package cn.zhang.violet.report


import cn.zhang.violet.bean.RptBussiness
import cn.zhang.violet.config.ConfigHandler
import com.google.gson.Gson
import org.apache.spark.sql.{SQLContext, SaveMode}
import org.apache.spark.{SparkConf, SparkContext}

object TerminalBussinessCore {


  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
    conf.setAppName("终端设备-运营商数据")
    conf.setMaster("local[*]")
    //设置spark程序采用的是序列化方式
    conf.set("spark.serializer","org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(conf)

    val sQLContext = new SQLContext(sc)
    //读取数据
    val dataFrame = sQLContext.read.parquet(ConfigHandler.parquetFilePath)
    //需要导入隐式转换
    import sQLContext.implicits._


    dataFrame.map(str => {
      val ispid = str.getAs[Int]("ispid")
      val ispname = str.getAs[String]("ispname")

      // 提取个指标相关的字段
      val reqMode = str.getAs[Int]("requestmode")
      val proNode = str.getAs[Int]("processnode")
      val effTive = str.getAs[Int]("iseffective")
      val billing = str.getAs[Int]("isbilling")
      val isbid = str.getAs[Int]("isbid")
      val iswin = str.getAs[Int]("iswin")
      val adOrderId = str.getAs[Int]("adorderid")

      val rawReq = if(reqMode == 1 && proNode >= 1) 1 else 0
      val effReq = if(reqMode == 1 && proNode >= 2) 1 else 0
      val adReq =if(reqMode == 1 && proNode >= 3) 1 else 0

      val adRtbReq = if(effTive == 1 && billing == 1 && isbid == 1 && adOrderId != 0) 1 else 0
      val adSuccRtbAndCostAndConsumption = if(effTive == 1 && billing == 1 && iswin == 1){

        val winPrice = str.getAs[Double]("winprice")
        val adPayment = str.getAs[Double]("adpayment")
        (1,adPayment /1000,winPrice /1000)

      } else (0,0d,0d)
      val adShow = if(reqMode == 2 && effTive == 1) 1 else 0
      val adClick = if(reqMode == 3 && effTive == 1) 1 else 0


      //组成一个对偶元祖
      ((ispid,ispname),List(rawReq, effReq, adReq, adRtbReq, adSuccRtbAndCostAndConsumption._1,
        adShow, adClick, adSuccRtbAndCostAndConsumption._2, adSuccRtbAndCostAndConsumption._3))
    }).reduceByKey((a,b) =>a.zip(b).map(t => t._1 + t._2))
      .map(tp => RptBussiness (
        tp._1._1,
        tp._1._2,
        tp._2(0).toInt,
        tp._2(1).toInt,
        tp._2(2).toInt,
        tp._2(3).toInt,
        tp._2(4).toInt,
        tp._2(5).toInt,
        tp._2(6).toInt,
        tp._2(7),
        tp._2(8)
      )).toDF().write.mode(SaveMode.Overwrite)
      .jdbc(ConfigHandler.url, "violet_prodata_Bussiness", ConfigHandler.props)


    sc.stop()
  }
}
